package org.hornetq.example.client;

import java.io.InputStream;
import java.io.Serializable;
import java.util.Properties;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;
import javax.naming.InitialContext;
import javax.naming.NamingException;

public class JmsHronetQUtil {

	private static final ThreadLocal<Session> threadLocal = new ThreadLocal<Session>();
	private static boolean useTopic = true;
	private static String queueName = "exampleTopic";

	private static JmsHronetQUtil uniqueInstance = null;

	public static Destination createDestination() throws NamingException {
		return createDestination(getQueueName());
	}

	private static InitialContext initialContext = null;

	static Connection connection = null;

	private JmsHronetQUtil() {
		try {
			initialContext = getContext();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	public static JmsHronetQUtil getInstance() {
		if (uniqueInstance == null) {
			uniqueInstance = new JmsHronetQUtil();
		}
		return uniqueInstance;
	}

	private InitialContext getContext() throws Exception {
		InputStream inStream = this.getClass().getClassLoader().getResourceAsStream("client-jndi.properties");
		Properties props = new Properties();
		try {
			props.load(inStream);
		} finally {
			if (inStream != null) {
				inStream.close();
			}
		}
		return new InitialContext(props);
	}

	public Session getSession(boolean transacted, int ackMode) throws JMSException, NamingException {
		Session session = (Session) threadLocal.get();
		if (session == null) {
			ConnectionFactory cf = (ConnectionFactory) initialContext.lookup("/ConnectionFactory");
			connection = cf.createConnection();
			session = connection.createSession(transacted, ackMode);
			threadLocal.set(session);
		}
		return session;
	}

	public Session getSession() throws JMSException, NamingException {
		return getSession(false, Session.AUTO_ACKNOWLEDGE);
	}

	public static void closeSession() throws JMSException {
		Session session = (Session) threadLocal.get();
		threadLocal.remove();
		if (session != null) {
			session.close();
		}
	}

	public static String getQueueName() {
		return queueName;
	}

	public static Destination createDestination(String subject) throws NamingException {
		// TODO Auto-generated method stub
		if (useTopic) {
			Topic topic = (Topic) initialContext.lookup(subject);
			return topic;
		} else {
			Queue queue = (Queue) initialContext.lookup(subject);
			return queue;
		}
	}

	public void sendMessage(Message m) throws Exception {
		sendMessage(m, queueName);
	}

	public void sendMessage(Serializable obj) throws Exception {
		sendMessage(obj, queueName);
	}

	public void sendMessage(Serializable obj, String queue) throws Exception {
		try {
			Session session = getSession();
			Destination destination = session.createTopic(queue);
			MessageProducer producer = session.createProducer(destination);
			TextMessage message = session.createTextMessage(obj.toString());
			producer.send(message);
		} finally {
			// stopConnection();
		}
	}

	public void sendMessage(Message m, String queue) throws Exception, JMSException {
		try {
			// startConnection();
			Session session = getSession();
			Destination destination = session.createTopic(queue);
			MessageProducer producer = session.createProducer(destination);
			producer.send(m);
		} finally {
			// stopConnection();
		}
	}

}
